package org.john.storm;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.Utils;

/**
 * Created by adsame on 17-1-26.
 */
public class StormTest {

    public static void main(String args[]) throws Exception{
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("spout", new TestSpout(), 3);
        builder.setBolt("bolt", new TestBolt(), 3).shuffleGrouping("spout");
        builder.setBolt("bolt2", new TestBolt(), 2).shuffleGrouping("bolt").shuffleGrouping("spout");
        StormSubmitter.submitTopology("test", new Config(), builder.createTopology());

        Config conf = new Config();
        conf.setDebug(true);
        conf.setNumWorkers(2);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("test", conf, builder.createTopology());
        Utils.sleep(10000);
        cluster.killTopology("test");
        cluster.shutdown();
    }
}
